home *** CD-ROM | disk | FTP | other *** search
/ Chip 2007 January, February, March & April / Chip-Cover-CD-2007-02.iso / Pakiet bezpieczenstwa / mini Pentoo LiveCD 2006.1 / mpentoo-2006.1.iso / livecd.squashfs / usr / lib / python2.4 / test / test_threadsignals.pyo (.txt) < prev    next >
Encoding:
Python Compiled Bytecode  |  2005-10-18  |  2.9 KB  |  87 lines
  1. # Source Generated with Decompyle++
  2. # File: in.pyo (Python 2.4)
  3.  
  4. '''PyUnit testing that threads honor our signal semantics'''
  5. import unittest
  6. import thread
  7. import signal
  8. import os
  9. import sys
  10. from test.test_support import run_unittest, TestSkipped
  11. if sys.platform[:3] in ('win', 'os2') or sys.platform == 'riscos':
  12.     raise TestSkipped, "Can't test signal on %s" % sys.platform
  13.  
  14. process_pid = os.getpid()
  15. signalled_all = thread.allocate_lock()
  16.  
  17. def registerSignals(.0):
  18.     (for_usr1, for_usr2, for_alrm) = .0
  19.     usr1 = signal.signal(signal.SIGUSR1, for_usr1)
  20.     usr2 = signal.signal(signal.SIGUSR2, for_usr2)
  21.     alrm = signal.signal(signal.SIGALRM, for_alrm)
  22.     return (usr1, usr2, alrm)
  23.  
  24.  
  25. def handle_signals(sig, frame):
  26.     signal_blackboard[sig]['tripped'] += 1
  27.     signal_blackboard[sig]['tripped_by'] = thread.get_ident()
  28.  
  29.  
  30. def send_signals():
  31.     os.kill(process_pid, signal.SIGUSR1)
  32.     os.kill(process_pid, signal.SIGUSR2)
  33.     signalled_all.release()
  34.  
  35.  
  36. class ThreadSignals(unittest.TestCase):
  37.     '''Test signal handling semantics of threads.
  38.        We spawn a thread, have the thread send two signals, and
  39.        wait for it to finish. Check that we got both signals
  40.        and that they were run by the main thread.
  41.     '''
  42.     
  43.     def test_signals(self):
  44.         signalled_all.acquire()
  45.         self.spawnSignallingThread()
  46.         signalled_all.acquire()
  47.         if signal_blackboard[signal.SIGUSR2]['tripped'] == 0 or signal_blackboard[signal.SIGUSR2]['tripped'] == 0:
  48.             signal.alarm(1)
  49.             signal.pause()
  50.             signal.alarm(0)
  51.         
  52.         self.assertEqual(signal_blackboard[signal.SIGUSR1]['tripped'], 1)
  53.         self.assertEqual(signal_blackboard[signal.SIGUSR1]['tripped_by'], thread.get_ident())
  54.         self.assertEqual(signal_blackboard[signal.SIGUSR2]['tripped'], 1)
  55.         self.assertEqual(signal_blackboard[signal.SIGUSR2]['tripped_by'], thread.get_ident())
  56.         signalled_all.release()
  57.  
  58.     
  59.     def spawnSignallingThread(self):
  60.         thread.start_new_thread(send_signals, ())
  61.  
  62.  
  63.  
  64. def test_main():
  65.     global signal_blackboard
  66.     signal_blackboard = {
  67.         signal.SIGUSR1: {
  68.             'tripped': 0,
  69.             'tripped_by': 0 },
  70.         signal.SIGUSR2: {
  71.             'tripped': 0,
  72.             'tripped_by': 0 },
  73.         signal.SIGALRM: {
  74.             'tripped': 0,
  75.             'tripped_by': 0 } }
  76.     oldsigs = registerSignals((handle_signals, handle_signals, handle_signals))
  77.     
  78.     try:
  79.         run_unittest(ThreadSignals)
  80.     finally:
  81.         registerSignals(oldsigs)
  82.  
  83.  
  84. if __name__ == '__main__':
  85.     test_main()
  86.  
  87.